package InitialRank;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

import mapred.AbstractReducer;
import mapred.util.Bytes;

public class InitialGetReady_Reducer extends AbstractReducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	@Override
	public void reduce(BytesWritable key, Iterator<BytesWritable> values,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		double initialRank = 1d;
		byte [] link = {};
		while(values.hasNext()) {
			ReadBytes v = new ReadBytes(values.next());
			if(v.size() == 8) {
				initialRank = v.nextDouble();
			} else {
				link = v.nextBytes(v.size());
			}
		}
		BytesWritable bw = new BytesWritable(Bytes.concat(Bytes.Double2Bytes(initialRank), link));
		output.collect(key, bw);
	}

}
